package com.atguigu.sparkstreaming.apps

import java.util

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.atguigu.realtime.constants.{PrefixConstant, TopicConstant}
import com.atguigu.realtime.utils.{KafkaProducerUtil, RedisUtil}
import com.atguigu.sparkstreaming.beans.StartLog
import com.atguigu.sparkstreaming.utils.{DStreamUtil, DateParseUtil}
import com.google.gson.Gson
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**

 *    at least once + 幂等输出(hbase)
 *
 *    柯里化:   函数   func()()
 *              定义一个函数，这个函数需要传入一个高阶函数，高阶函数拥有多个参数列表，称为函数柯里化
 *
 *     闭包：   在一个JVM中使用到了此JVM没有定义的变量。
 *              具体spark中:  在rdd的算子中使用了算子方法外定义的变量。
 *                          rdd的算子是在Executor的JVM中运行。
 *                          使用的算子外定义的变量在Driver端定义。
 *
 *
 *       最后数据是什么样取决于最后一个 RDD[T] ,T
 *
 *
 *       输入：    ds: InputDStream[ConsumerRecord[String, String]]
 *                  DStream是RDD的无限集合。
 *                  RDD[ConsumerRecord[String, String]]  --->map ---->RDD[StartLog] ---> filter ---->filter--->RDD[StartLog]
 *                  ----写入hbase
 *
 *
 *
 */
object DAUApp extends BaseApp {
  override var groupId: String = "realtime220309"
  override var topic: String = TopicConstant.STARTUP_LOG
  override var appName: String = "DAUApp"
  override var batchDuration: Int = 10


  /*
      fastjon的封装规则:
          要求封装的jsonstr 和 javabean的模式要一一对应。
            jsonstr有几个一级属性，javabean有和它同样的一级属性。属性名必须一致!

      数据演示:
          当前这批:  [
          {(mid1,2022-07-19),log1}
          {(mid1,2022-07-19),log2}
          {(mid1,2022-07-19),log3}
          {(mid2,2022-07-19),log4}
                  ]

             分组后:
           [
            ((mid1, 2022-07-19), Iterable[log1,log2,log3])
            ((mid2, 2022-07-19), Iterable[log4])
           ]

            flatMap:
           map:  Iterable[log1,log2,log3] ---> List[log1,log2,log3] ---> List[log2,log1,log3] --->List[log2]
           flatten:  List[log2] ----> log2

            Iterable[log4] ----> log4

   */
  def parseBean(rdd: RDD[ConsumerRecord[String, String]]):RDD[StartLog] = {

    rdd.map(record => {

      val startLog: StartLog = JSON.parseObject(record.value(), classOf[StartLog])

      //继续封装id,starttime,startdate
      startLog.start_time = DateParseUtil.parseMillTsToDateTime(startLog.ts.toLong)
      startLog.start_date = DateParseUtil.parseMillTsToDate(startLog.ts.toLong)
      startLog.id = startLog.start_time + "_" + startLog.mid

      startLog

    })

  }

  /*
      groupByKey： 将map阶段的所有的数据，都shuffle。
                  分组!

       当前的需求是分组聚合(reduceByKey)！
            一个组 --------------> 一条数据

       reduceByKey： 可以map端聚合！
   */
  def removeDulicateLogInCommonBatch(rdd: RDD[StartLog]):RDD[StartLog] = {

    val rdd1: RDD[((String, String), StartLog)] = rdd.map(log => ((log.mid, log.start_date), log))
   /* val rdd2: RDD[((String, String), Iterable[StartLog])] = rdd1.groupByKey()
    val rdd3: RDD[StartLog] = rdd2.flatMap {
      case ((mid, date), logs) => logs.toList.sortBy(_.ts).take(1)
    }*/


    val rdd2: RDD[((String, String), StartLog)] = rdd1.reduceByKey((log1, log2) => {

      if (log1.ts < log2.ts) {
        log1
      } else {
        log2
      }

    })
    rdd2.values
  }

  /*
      设计数据在redis中存储的K-V
        数据:  日期(哪一天)   设备id(哪些设备已经记录过最早启动日志了)

        K：
            字符串，唯一标识
        V：
            单值: 1:1 string,hash,
           集合: 1:N list,set,zset

       参考粒度:
                  1天的1个设备是1个k-v:
                       K: 日期_设备id
                       v:  随便

                  1天的N个设备是1个k-v:
                      k: 日期
                      v: Set(mid)

       使用场景:  从一批数据中判断当前这批数据，哪些mid已经在redis中存在!
                 是否方便管理的角度:   1天的N个设备是1个k-v

   */
  def removeDulicateLogInHistoryBatch(rdd: RDD[StartLog]):RDD[StartLog]  = {

    rdd.mapPartitions(partition => {

      val jedis: Jedis = RedisUtil.getJedis

      val filterdLogs: Iterator[StartLog] = partition.filter(log => !jedis.sismember(PrefixConstant.dau_redis_Preffix + log.start_date, log.mid))

      jedis.close()

      filterdLogs

    })

  }

  def main(args: Array[String]): Unit = {

    //重写context
    context = new StreamingContext("local[*]",appName,Seconds(batchDuration))

    runApp{

      val ds: InputDStream[ConsumerRecord[String, String]] = DStreamUtil.createDStream(groupId, context, topic)

      ds.foreachRDD(rdd => {

        if (!rdd.isEmpty()){

          //获取当前消费到的这个批次偏移量
          val ranges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

          //封装样例类
          val rdd1: RDD[StartLog] = parseBean(rdd)
          
          //同批次去重
          val rdd2: RDD[StartLog] = removeDulicateLogInCommonBatch(rdd1)

          //历史批次去重
          val rdd3: RDD[StartLog] = removeDulicateLogInHistoryBatch(rdd2)

          //导入phoenix-spark提供方法
          import org.apache.phoenix.spark._

          rdd3.cache()

          println("即将写入:"+rdd3.count())
          /*
            tableName: 表名,
            cols: Seq[String],  Seq(序列，有顺序的集合)。
                    指当前RDD中的数据类型(StartLog)的哪些列，按照顺序要写入 对应表的哪些列

            conf: Configuration = new Configuration,
                      HBaseConfiguration.create() : 先new Configuration，再读取hbase-site.xml和hbase-default.xml

            zkUrl: Option[String] = hbase 用的zk的地址
           */
          //写入hbase
          rdd3.saveToPhoenix("REALTIME2022_STARTLOG",
            Seq("ID","OPEN_AD_MS","OS","CH","IS_NEW","MID","OPEN_AD_ID","VC","AR",
              "UID","ENTRY","OPEN_AD_SKIP_MS","MD","LOADING_TIME","BA","TS","START_DATE","START_TIME"),
            HBaseConfiguration.create(),
            Some("hadoop102:2181")
          )

          //写入redis
          rdd3.foreachPartition(partition => {

            val jedis: Jedis = RedisUtil.getJedis

            partition.foreach(log => jedis.sadd(PrefixConstant.dau_redis_Preffix + log.start_date, log.mid))

            jedis.close()

          })

          //提交offsets
          ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)

        }


      })


    }


  }
}
